package com.ds.lens.flink.monitor.common;

import com.ds.lens.flink.monitor.MonitorJobContext;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.util.serialization.KeyedDeserializationSchema;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;

@Slf4j
public class CommonDeserializationSchema<T> implements KeyedDeserializationSchema<T> {

    private final Class<T> classT;

    public CommonDeserializationSchema(Class<T> classT) {
        this.classT = classT;
    }

    @Override
    public T deserialize(byte[] messageKey, byte[] message, String topic, int partition, long offset) throws IOException {
        try {
            return MonitorJobContext.getInstance().getGson()
                    .fromJson(new InputStreamReader(new ByteArrayInputStream(message), StandardCharsets.UTF_8), classT);
        } catch (Exception e) {
            log.error("Failed to parse JSON", e);
        }
        return null;
    }

    @Override
    public boolean isEndOfStream(T nextElement) {
        return false;
    }

    @Override
    public TypeInformation<T> getProducedType() {
        return TypeInformation.of(classT);
    }
}
